package org.common;

import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import java.io.IOException;
import java.util.*;

/**
 * 学生表操作类
 * 实现对学生表的具体数据操作
 * 编写人：张凯旋
 * 编写日期：2025年4月28日
 */
public class Student {
    private static final String TABLE_NAME = "student";
    private static final String INFO_FAMILY = "stu_info";
    private final MyHBase hbase;

    public Student() {
        this.hbase = new MyHBase();
    }

    /**
     * 查询年龄≥18且姓名包含"cong"的学生
     * @throws IOException 如果查询失败
     */
    public void findStudentsByAgeAndName() throws IOException {
        try (Table table = hbase.getConnection().getTable(TableName.valueOf(TABLE_NAME))) {
            // 创建组合过滤器
            FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);

            // 年龄≥18的过滤器
            SingleColumnValueFilter ageFilter = new SingleColumnValueFilter(
                    Bytes.toBytes(INFO_FAMILY),
                    Bytes.toBytes("age"),
                    CompareFilter.CompareOp.GREATER_OR_EQUAL,
                    Bytes.toBytes("18")
            );
            filterList.addFilter(ageFilter);

            // 姓名包含"cong"的过滤器
            SingleColumnValueFilter nameFilter = new SingleColumnValueFilter(
                    Bytes.toBytes(INFO_FAMILY),
                    Bytes.toBytes("name"),
                    CompareFilter.CompareOp.EQUAL,
                    new SubstringComparator("cong")
            );
            filterList.addFilter(nameFilter);

            // 创建扫描对象并设置过滤器
            Scan scan = new Scan();
            scan.setFilter(filterList);

            // 执行查询
            ResultScanner scanner = table.getScanner(scan);
            System.out.println("年龄≥18且姓名包含'cong'的学生：");
            for (Result result : scanner) {
                printResult(result);
            }
            scanner.close();
        }
    }

    /**
     * 扫描全表数据
     * @throws IOException 如果扫描失败
     */
    public void scanAllStudents() throws IOException {
        try (Table table = hbase.getConnection().getTable(TableName.valueOf(TABLE_NAME))) {
            Scan scan = new Scan();
            ResultScanner scanner = table.getScanner(scan);
            System.out.println("学生表全部信息：");
            for (Result result : scanner) {
                printResult(result);
            }
            scanner.close();
        }
    }

    /**
     * 扫描指定行键范围的数据
     * @throws IOException 如果扫描失败
     */
    public void scanRangeStudents() throws IOException {
        try (Table table = hbase.getConnection().getTable(TableName.valueOf(TABLE_NAME))) {
            Scan scan = new Scan();
            scan.setStartRow(Bytes.toBytes("S0003"));
            scan.setStopRow(Bytes.toBytes("S0006")); // S0005之后的下一个行键
            ResultScanner scanner = table.getScanner(scan);
            System.out.println("行键范围S0003到S0005的学生：");
            for (Result result : scanner) {
                printResult(result);
            }
            scanner.close();
        }
    }

    /**
     * 扫描指定列族的所有name列
     * @throws IOException 如果扫描失败
     */
    public void scanNameColumn() throws IOException {
        try (Table table = hbase.getConnection().getTable(TableName.valueOf(TABLE_NAME))) {
            Scan scan = new Scan();
            scan.addColumn(Bytes.toBytes(INFO_FAMILY), Bytes.toBytes("name"));
            ResultScanner scanner = table.getScanner(scan);
            System.out.println("所有学生的姓名：");
            for (Result result : scanner) {
                String rowKey = Bytes.toString(result.getRow());
                String name = Bytes.toString(result.getValue(Bytes.toBytes(INFO_FAMILY), Bytes.toBytes("name")));
                System.out.println("行键: " + rowKey + ", 姓名: " + name);
            }
            scanner.close();
        }
    }

    /**
     * 统计每位创建者的学生数量
     * @throws IOException 如果统计失败
     */
    public void countStudentsByAuthor() throws IOException {
        Map<String, Integer> authorCount = new HashMap<>();

        try (Table table = hbase.getConnection().getTable(TableName.valueOf(TABLE_NAME))) {
            Scan scan = new Scan();
            scan.addColumn(Bytes.toBytes(INFO_FAMILY), Bytes.toBytes("createAuthor"));
            ResultScanner scanner = table.getScanner(scan);

            for (Result result : scanner) {
                String author = Bytes.toString(result.getValue(Bytes.toBytes(INFO_FAMILY), Bytes.toBytes("createAuthor")));
                authorCount.put(author, authorCount.getOrDefault(author, 0) + 1);
            }
            scanner.close();
        }

        System.out.println("每位创建者的学生数量：");
        for (Map.Entry<String, Integer> entry : authorCount.entrySet()) {
            System.out.println("创建者: " + entry.getKey() + ", 数量: " + entry.getValue());
        }
    }

    /**
     * 打印查询结果
     * @param result 查询结果
     */
    private void printResult(Result result) {
        String rowKey = Bytes.toString(result.getRow());
        String name = Bytes.toString(result.getValue(Bytes.toBytes(INFO_FAMILY), Bytes.toBytes("name")));
        String age = Bytes.toString(result.getValue(Bytes.toBytes(INFO_FAMILY), Bytes.toBytes("age")));
        String admissionTime = Bytes.toString(result.getValue(Bytes.toBytes(INFO_FAMILY), Bytes.toBytes("AdmissionTime")));
        String createAuthor = Bytes.toString(result.getValue(Bytes.toBytes(INFO_FAMILY), Bytes.toBytes("createAuthor")));

        System.out.println(String.format("行键: %s, 姓名: %s, 年龄: %s, 入学时间: %s, 创建者: %s",
                rowKey, name, age, admissionTime, createAuthor));
    }

    /**
     * 关闭HBase连接
     * @throws IOException 如果关闭失败
     */
    public void close() throws IOException {
        hbase.close();
    }
}